package com.zhb.delayQueue.queueOp;

import org.redisson.api.RMap;
import org.redisson.api.RScoredSortedSet;
import org.redisson.api.RedissonClient;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by millions on 2017/1/10.
 * Redisson实现的Operation
 */
public class RedissonOperation implements Operation {

    /**
     * redisson操作client
     */
    private final RedissonClient client;

    /**
     * 存放jobId的unsorted set名称
     */
    private final ArrayList<String> delayBuckets;

    /**
     * 下一个存放的bucket index
     */
    private AtomicLong nextBucketIndex = new AtomicLong(0L);

    public RedissonOperation(RedissonClient client, ArrayList<String> delayBuckets) {
        this.client = client;
        this.delayBuckets = delayBuckets;
    }


    @Override
    public boolean enqueue(String jobId, String topic, int ttr, String body, long delay) {
        RMap<String, String> map = client.getMap(jobId);
        HashMap<String, String> jobData = new HashMap<>();
        jobData.put("topic", topic);
        jobData.put("ttr", String.valueOf(ttr));
        jobData.put("body", body);
        jobData.put("delay", String.valueOf(delay));
        map.putAll(jobData);
        if (map.size() > 0) {
            //delay bucket size 正常情况下肯定小于MAX INTEGER
            int bucketIndex = (int) (nextBucketIndex.incrementAndGet() % delayBuckets.size());
            RScoredSortedSet<String> bucket = client.getScoredSortedSet(delayBuckets.get(bucketIndex));
            double score = System.currentTimeMillis() / 1000 + delay;
            return bucket.add(score, jobId);
        }
        return false;
    }

    @Override
    public boolean ready(String jobId, String topic) {
        return false;
    }

    @Override
    public boolean reserve(String jobId, long ttr) {
        return false;
    }

    @Override
    public boolean delete(String jobId) {
        return false;
    }
}
